import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

/**
 * 1、找到ip所属区域
 * http.log：用户访问网站所产生的日志。
 *          日志格式为：时间戳、IP地址、访问网址、访问数据、浏览器信息等
 * ip.dat：ip段数据，记录着一些ip段范围对应的位置
 * 文件位置：data/http.log、data/ip.dat
 * 要求：将 http.log 文件中的 ip 转换为地址。如将 122.228.96.111 转为温州，并统计各城市的总访问量
 */

object Subject1 {
  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.WARN)
    val spark = SparkSession
      .builder()
      .appName(this.getClass.getCanonicalName)
      .master("local[*]")
      .getOrCreate()
    import spark.implicits._
    import spark._

    //读ip.dat，获得ip段数据和城市名
    val ipData: Array[(Long, Long, String)] = spark.read
      //分隔符为|
      .option("delimiter", "|")
      .csv("data/ip.dat")
      //取出ip段和对应城市，数据中已有转换为Long类型的ip数据，只需要将读出的字符串类型转为Long类型
      .map(row => (row.getString(2).toLong, row.getString(3).toLong, row.getString(7)))
      .collect()
      //根据ip段起始地址排序，方便之后的查找
      .sortBy(_._1)

    ipData.take(10).foreach(println)

    println("************************************")

    //读取访问日志信息
    spark.read
      .option("delimiter", "|")
      .csv("data/http.log")
      //只取出第二列数据，ip地址
      .map(row => row.getString(1))
      //更改列名
      .withColumnRenamed("value", "ip")
      //创建临时表t1
      .createTempView("t1")

    sql(
      """
        |select * from t1 limit 10
        |""".stripMargin
    ).show

    //表t1中是点分十进制表示的ip地址，需要转化为Long类型
    def ipToLong(ip: String): Long = {
      ip.split("\\.").map(_.toLong).reduce((x, y) => x << 8 | y)
    }

    //根据传入的ip地址确定对应的城市名，使用二分查找
    def getCity(ip: Long): String = {
      var left = 0
      var right = ipData.length - 1
      var mid = 0
      //flag用来停止循环
      var flag = true
      var result: String = null
      while (left <= right && flag) {
        mid = (left + right) / 2
        if (ip >= ipData(mid)._1 && ip <= ipData(mid)._2) {
          flag = false
          result = ipData(mid)._3
        }
        else if (ip < ipData(mid)._1)
          right = mid - 1
        else
          left = mid + 1
      }
      result
    }

    //注册udf函数
    spark.udf.register("ipToLong", ipToLong _)
    spark.udf.register("getCity", getCity _)

    //t1中只有一列value，表示点分十进制的ip地址，调用ipToLong()转为Long类型后传入getCity()得到对应城市名city
    //根据city分组，统计各城市的总访问量
    sql(
      """
        |select getCity(ipToLong(ip)) as city, count(1) as count
        |  from t1
        |group by getCity(ipToLong(ip))
        |""".stripMargin
    ).show

    spark.close()
  }
}
